#!/usr/bin/python3
# -*- coding: utf-8 -*-

"""
# @version: v1.0
# @author : cd
# @Email : 19688513@qq.com
# @Project : new-horizons-engine
# @File : SignalGenerator.py
# @Software: PyCharm
# @time: 2025/6/4 10:18
# @description : 信号生成器类，负责生成交易信号
"""


import urllib3
import logging
from typing import List, Tuple, Dict, Any, Optional
import pandas as pd
from datetime import datetime

# 配置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
# 禁用不安全的请求警告
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)


class SignalGenerator:
    """信号生成器类，负责生成交易信号"""

    def __init__(self):
        # 定义信号名称和编号的映射
        self.SIGNAL_MAP = {

            'signal_vol': 1,
            'signal_price': 2,
            'signal_vol_break': 3,
            'signal_trend': 4,
            'signal_volume_price': 5,
            'signal_volatility': 6,
            'signal_rsi': 7,
            'signal_vol_macd': 8,
            'signal_vol_rsi': 9,
            'signal_vol_boll': 10,
            'signal_vol_kdj': 11
        }

        # 信号描述
        self.SIGNAL_DESCRIPTIONS = {
            1: "成交量双均线条件：成交量接近5日均量且低于60日均量的70%",
            2: "价格均线多头排列：5日>10日>20日均线，且5日均线上扬",
            3: "量能趋势突破：5日成交量均线上穿20日成交量均线",
            4: "趋势延续验证：成交量均线多头排列(5>20>60)且价格高于3日均价",
            5: "量价齐升：成交量放大10%以上，阳线且突破前日高点",
            6: "波动率适中：14日ATR处于近1年30%-70%分位之间",
            7: "价格RSI超卖：价格RSI低于其14日均值减1.5倍标准差",
            8: "成交量MACD金叉：成交量MACD上穿信号线",
            9: "成交量RSI超卖：成交量RSI < 30",
            10: "成交量布林带收窄：带宽小于20日均宽的70%，且成交量在通道内",
            11: "成交量KDJ金叉：K线上穿D线且J值>20",
            12: "量价共振：量价齐升+成交量MACD金叉+价格RSI>50",
            13: "底部反转信号：成交量RSI超卖+成交量KDJ金叉+价格RSI<40"
        }

    def generate_signals(self, df: pd.DataFrame) -> Dict[str, Any]:
        """生成买入和卖出信号，返回结构化信号信息"""
        if df.empty:
            return {
                'signals_df': pd.DataFrame(),
                'signals_list': [],
                'signals_summary': []
            }

        signals = df.copy()
        signals['price'] = df['close']
        signals['volume'] = (df['volume'] / 10000).round(2)  # 转换为万手

        # 计算价格RSI
        if 'price_rsi' not in signals.columns:
            signals['price_rsi'] = self._calculate_rsi(df['close'])

        # 原有信号规则
        signals = self._generate_basic_signals(signals)

        # 新增基于成交量技术指标的信号规则
        signals = self._generate_volume_signals(signals)

        # 复合信号规则
        signals = self._generate_composite_signals(signals)

        # 创建信号输出
        signals, signal_output, signals_summary = self._create_signal_output(signals)

        return {
            'signals_df': signals,
            'signals_list': signal_output,
            'signals_summary': signals_summary
        }

    def _calculate_rsi(self, series: pd.Series, period: int = 14) -> pd.Series:
        """计算相对强弱指数(RSI)"""
        delta = series.diff()
        gain = (delta.where(delta > 0, 0)).fillna(0)
        loss = (-delta.where(delta < 0, 0)).fillna(0)

        # 使用简单移动平均
        avg_gain = gain.rolling(window=period, min_periods=period).mean()
        avg_loss = loss.rolling(window=period, min_periods=period).mean()

        # 计算RS
        rs = avg_gain / (avg_loss + 1e-8)  # 避免除零
        return 100 - (100 / (1 + rs))

    def _generate_basic_signals(self, signals: pd.DataFrame) -> pd.DataFrame:
        """生成基础信号规则"""
        # 信号1：成交量双均线条件
        signals['signal_vol'] = (
                (abs(signals['volume'] - signals['vol_ma5']) / signals['vol_ma5'].clip(lower=1e-6) < 0.02)
                &
                (signals['volume'] < signals['vol_ma60'] * 0.7)
        ).fillna(False)

        # 信号2：价格均线多头排列
        signals['ma5_slope'] = signals['ma5'].diff(3) > 0
        signals['signal_price'] = (
                (signals['ma5'] > signals['ma10']) &
                (signals['ma10'] > signals['ma20']) &
                (signals['close'] > signals['ma20']) &
                signals['ma5_slope']
        ).fillna(False)

        # 信号3：量能趋势突破
        signals['signal_vol_break'] = (
                (signals['vol_ma5'] > signals['vol_ma20']) &
                (signals['vol_ma5'].shift(1) <= signals['vol_ma20'].shift(1)) &
                (signals['vol_ma5'] > signals['vol_ma5'].shift(3))
        ).fillna(False)

        # 信号4：趋势延续验证
        signals['signal_trend'] = (
                (signals['vol_ma5'] > signals['vol_ma20']) &
                (signals['vol_ma20'] > signals['vol_ma60']) &
                (signals['close'] > signals['close'].rolling(3).mean())
        ).fillna(False)

        # 信号5：量价齐升
        signals['signal_volume_price'] = (
                (signals['volume'] > signals['vol_ma5'] * 1.1) &
                (signals['close'] > signals['open']) &
                (signals['close'] > signals['high'].shift(1))
        ).fillna(False)

        # 信号6：波动率过滤
        atr = (signals['high'] - signals['low']).rolling(14).mean()
        q30 = atr.rolling(252).quantile(0.3)
        q70 = atr.rolling(252).quantile(0.7)
        signals['signal_volatility'] = atr.between(q30, q70).astype(bool)

        # 信号7：RSI动态超卖
        rsi_ma = signals['price_rsi'].rolling(14).mean()
        rsi_std = signals['price_rsi'].rolling(14).std()
        signals['signal_rsi'] = (
                signals['price_rsi'] < (rsi_ma - 1.5 * rsi_std)
        ).fillna(False)

        return signals

    def _generate_volume_signals(self, signals: pd.DataFrame) -> pd.DataFrame:
        """生成基于成交量指标的信号"""
        # 信号8：成交量MACD金叉
        signals['signal_vol_macd'] = (
                (signals['vol_macd'] > signals['vol_signal']) &
                (signals['vol_macd'].shift(1) <= signals['vol_signal'].shift(1))
        ).fillna(False)

        # 信号9：成交量RSI超卖
        signals['signal_vol_rsi'] = (signals['vol_rsi'] < 30).fillna(False)

        # 信号10：成交量布林带收窄（波动率低）
        if 'vol_boll_upper' in signals.columns and 'vol_boll_lower' in signals.columns:
            boll_width = signals['vol_boll_upper'] - signals['vol_boll_lower']
            avg_boll_width = boll_width.rolling(20).mean()
            signals['signal_vol_boll'] = (
                    (boll_width < avg_boll_width * 0.7) &  # 带宽小于平均宽度的70%
                    (signals['volume_wan'] > signals['vol_boll_lower']) &  # 成交量在布林带范围内
                    (signals['volume_wan'] < signals['vol_boll_upper'])
            ).fillna(False)
        else:
            signals['signal_vol_boll'] = False

        # 信号11：成交量KDJ金叉
        if 'vol_k' in signals.columns and 'vol_d' in signals.columns:
            signals['signal_vol_kdj'] = (
                    (signals['vol_k'] > signals['vol_d']) &
                    (signals['vol_k'].shift(1) <= signals['vol_d'].shift(1)) &
                    (signals['vol_j'] > 20)  # 避免在超卖区假信号
            ).fillna(False)
        else:
            signals['signal_vol_kdj'] = False

        return signals

    def _generate_composite_signals(self, signals: pd.DataFrame) -> pd.DataFrame:
        """生成复合信号规则"""
        # 信号12：量价共振（价格突破+成交量确认）
        signals['signal_combo_breakout'] = (
                signals['signal_volume_price'] &  # 量价齐升
                signals['signal_vol_macd'] &  # 成交量MACD金叉
                (signals['price_rsi'] > 50)  # 价格趋势中等强度
        ).fillna(False)
        self.SIGNAL_MAP['signal_combo_breakout'] = 12

        # 信号13：底部反转信号
        signals['signal_combo_bottom'] = (
                signals['signal_vol_rsi'] &  # 成交量RSI超卖
                signals['signal_vol_kdj'] &  # 成交量KDJ金叉
                (signals['price_rsi'] < 40)  # 价格RSI超卖
        ).fillna(False)
        self.SIGNAL_MAP['signal_combo_bottom'] = 13

        return signals

    def _create_signal_output(self, signals: pd.DataFrame) -> Tuple[pd.DataFrame, List[Dict], List[Dict]]:
        """创建信号输出结构"""
        signals['signals'] = ""
        signal_output = []

        # 确保存在日期列
        if 'date' not in signals.columns:
            # 如果索引是日期类型，则使用索引作为日期
            if isinstance(signals.index, pd.DatetimeIndex):
                signals['date'] = signals.index
            else:
                logging.error("DataFrame中缺少日期列且索引不是日期类型")
                return signals, [], []

        # 收集所有信号触发信息
        for signal_col, signal_id in self.SIGNAL_MAP.items():
            # 获取触发信号的日期
            trigger_mask = signals[signal_col].fillna(False)
            trigger_rows = signals[trigger_mask]

            # 为每个触发行创建详细记录
            for idx, row in trigger_rows.iterrows():
                # 获取该日期所有相关指标值
                record = {
                    'date': row['date'].strftime('%Y-%m-%d'),
                    'stock_code': row['symbol'],
                    'stock_name': row['stock_name'],
                    'rating': row['rating'],
                    'industry': row['industry'],
                    'change_pct': row['change_pct'],
                    'signal_id': signal_id,
                    'signal_name': signal_col,
                    'close_price': round(row['close'], 2),
                    'volume': round(row['volume'], 1),
                    'price_rsi': round(row['price_rsi'], 1) if 'price_rsi' in row else None,
                    'vol_rsi': round(row['vol_rsi'], 1) if 'vol_rsi' in row else None,
                    'vol_macd': round(row['vol_macd'], 3) if 'vol_macd' in row else None,
                    'vol_k': round(row['vol_k'], 1) if 'vol_k' in row else None,
                    'vol_d': round(row['vol_d'], 1) if 'vol_d' in row else None,
                    'description': self.SIGNAL_DESCRIPTIONS.get(signal_id, "未知信号")
                }
                signal_output.append(record)

            # 在DataFrame中标记信号
            signals.loc[trigger_mask, 'signals'] = signals.loc[trigger_mask, 'signals'].apply(
                lambda x: f"{x},{signal_id}" if x else str(signal_id))

        # 整理信号列格式
        signals['signals'] = signals['signals'].str.strip(',')

        # 生成信号统计摘要
        signals_summary = []
        for signal_col, signal_id in self.SIGNAL_MAP.items():
            trigger_count = signals[signal_col].sum()
            last_trigger_rows = signals[signals[signal_col]]

            signals_summary.append({
                'signal_id': signal_id,
                'signal_name': signal_col,
                'trigger_count': int(trigger_count),
                'last_trigger': last_trigger_rows['date'].max().strftime(
                    '%Y-%m-%d') if not last_trigger_rows.empty else None,
                'description': self.SIGNAL_DESCRIPTIONS.get(signal_id, "未知信号")
            })

        # 按信号ID排序
        signal_output.sort(key=lambda x: (x['date'], x['signal_id']))
        signals_summary.sort(key=lambda x: x['signal_id'])

        return signals, signal_output, signals_summary